#include <sstream>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>
#include <boost/thread/thread.hpp>
#include "TCPServer.h"
#include "../CommError.h"
#include "../../system/Observers.h"
#include <signal.h>

using namespace kc_sf;

namespace kc_sf
{
	bool SERV_END = false;
}

// ----------------------------------------------------------- //

TCPServer::TCPServer( int port, int max_connections, CommRequestHandler *client_handler )
: CommServer( client_handler ),
  tcp_port (port),
  tcp_listening_socket( -1 ),
  tcp_is_listening( false ),
  tcp_top_client_handler( client_handler )
{
}

// ----------------------------------------------------------- //

TCPServer::~TCPServer()
{
	tcp_client_threads.join_all();

	closeAllConnections ();
	if(tcp_top_client_handler)
	{
		delete tcp_top_client_handler;
		tcp_top_client_handler = NULL;
	}
}

// ----------------------------------------------------------- //

void TCPServer::startListening()
{
	// block signals
    sigset_t signal_set;
    sigemptyset(&signal_set);
    sigaddset(&signal_set, SIGINT);
    sigaddset(&signal_set, SIGTERM);
    sigaddset(&signal_set, SIGHUP);
    sigaddset(&signal_set, SIGPIPE); // http://www.unixguide.net/network/socketfaq/2.19.shtml
    pthread_sigmask(SIG_BLOCK, &signal_set, NULL);
    {
		boost::mutex::scoped_lock lock (tcp_listen_guard);
		if( tcp_is_listening )
			return;

		openListeningSocket();

		// nasluch
		if( listen( tcp_listening_socket, tcp_max_connections ) )
		{
			close( tcp_listening_socket );
			throw CommError( FailedInit, "TCPServer. Bląd listen." );
		}

		tcp_is_listening = true;
	}

	// glowna petla
	while( !SERV_END && tcp_is_listening )
    {
        // blokujacy accept <- obsluzyc sygnal przerwania
        int assoc_sock = acceptClient ();
        if( assoc_sock < 0 )
            continue;

		// dodaj gniazdko do listy gniazdek klientow
        {
			unsigned long cid = AllConnectionsInObserver.add() - 1;
			ConnectionsInObserver.add();
        	tcp_logger << detail << receiver << cidIn( cid ) << "Nowy klient nawiązał połączenie na porcie " << tcp_port << flush;

			boost::mutex::scoped_lock lock (tcp_clients_guard);
			Connection client( assoc_sock, cid );
			tcp_clients.push_back( client );

	    	// nowy watek przetwarzajacy - kopiuje client_handler
	    	boost::thread *handler = new boost::thread( &TCPServer::handleClient, this, client );
	    	tcp_client_threads.add_thread( handler );
        }
    }

    if( SERV_END )
        closeAllConnections();
}

// ----------------------------------------------------------- //

void TCPServer::stopListening()
{
	boost::mutex::scoped_lock lock( tcp_listen_guard );

    if( tcp_is_listening )
    {
    	tcp_is_listening = false;
    	close( tcp_listening_socket );

    	tcp_logger << detail << receiver << "Koniec nasłuchu" << flush;
    }
}

// ----------------------------------------------------------- //

std::list<int> TCPServer::getOpenConnectionSockets ()
{
	std::list<int> sockets;
	boost::mutex::scoped_lock lock( tcp_clients_guard );
	for( std::list<Connection>::const_iterator it = tcp_clients.begin();
			it != tcp_clients.end();
			++ it)
	{
		sockets.push_back( it->cSocketFd );
	}
	return sockets;
}

// ----------------------------------------------------------- //

void TCPServer::closeConnection ( int client_sock )
{
	boost::mutex::scoped_lock lock( tcp_clients_guard );

	interruptClientHandler( client_sock );
	close( client_sock );

	for( std::list<Connection>::iterator it = tcp_clients.begin();
			it != tcp_clients.end();
			++ it)
	{
		if( it->cSocketFd == client_sock )
		{
			unsigned long cid = it->cCid;
			tcp_clients.erase( it );
			tcp_logger << detail << receiver << cidIn( cid ) << "Zamknięto połączenie z klientem" << flush;
			break;
		}
	}
}

// ----------------------------------------------------------- //

void TCPServer::closeAllConnections ()
{
	tcp_client_threads.interrupt_all();
	boost::mutex::scoped_lock lock( tcp_clients_guard );
	for( std::list<Connection>::iterator it = tcp_clients.begin();
			it != tcp_clients.end();
			++ it)
	{
		interruptClientHandler( it->cSocketFd );
		close( it->cSocketFd );
	}
	tcp_clients.clear( );
	tcp_logger << detail << receiver << "Zamknięto wszystkie połączenia" << flush;
}

// ----------------------------------------------------------- //

void TCPServer::openListeningSocket()
{
	// tworzy gniazdo
	tcp_listening_socket = socket( AF_INET, SOCK_STREAM, 0 );
	if( tcp_listening_socket < 0 )
	{
		throw CommError( FailedInit,
				"TCPServer: Nie udało się otworzyć gniazda do nasłuchu" );
	}

	// ustawia opcje gniazda

	// nie czeka na zakończenie procesu zamykania połączenia z adresem
	int on = 1;                                        // wartosc nadawana opcji REUSEADDR
	setsockopt( tcp_listening_socket, SOL_SOCKET,             // poziom opcji = gniazdo
				SO_REUSEADDR,                          // bind nie bedzie spr czy adres jest w uzyciu
				( const char * )&on, sizeof( on ) );

	// po zamknięciu gniazda czeka 30 sekund, zeby nie przeslane
	// dane mialy szanse zostać przesłane
	struct linger linger = { 0 };

	linger.l_onoff = 1;         // wlacz utrzymywanie polaczenia
	linger.l_linger = 30;       // czas w s
	setsockopt( tcp_listening_socket, SOL_SOCKET,
				SO_LINGER,
				( const char * )&linger, sizeof( linger ) );

	// dowiazuje adres do gniazda
	struct sockaddr_in server = {0};
	server.sin_family = AF_INET;
	server.sin_addr.s_addr = inet_addr( "127.0.0.1" ); // adres IP hosta serwera - loopback
	server.sin_port = htons( tcp_port );      // konwersja na network byte order

	if( bind( tcp_listening_socket, (struct sockaddr *)&server, sizeof( server ) ) < 0 )
	{
		close( tcp_listening_socket );
		throw CommError( FailedInit,
				"TCPServer: Nie udało się związać gniazda z adresem i portem" );
	}
}

// ----------------------------------------------------------- //

int TCPServer::acceptClient ()
{
    struct sockaddr_in client = { 0 }; // dane klienta
    int sockaddr_in_size = sizeof( client );

	int assoc_sock = accept(
			tcp_listening_socket,
			( struct sockaddr * )&client,
			( socklen_t * )&sockaddr_in_size );

	if( assoc_sock <= 0 )
		throw CommError( FailedAccept,
				"TCPServer: Nie udało się zaakceptować połączenia od kienta (błąd accept)");

	return assoc_sock;
}

// ----------------------------------------------------------- //

void TCPServer::handleClient( Connection client )
{
	// block signals
    sigset_t signal_set;
    sigemptyset(&signal_set);
    sigaddset(&signal_set, SIGINT);
    sigaddset(&signal_set, SIGTERM);
    sigaddset(&signal_set, SIGHUP);
    sigaddset(&signal_set, SIGPIPE); // http://www.unixguide.net/network/socketfaq/2.19.shtml
    pthread_sigmask(SIG_BLOCK, &signal_set, NULL);

//	CommRequestHandler * client_handler = tcp_top_client_handler->create( client.cCid );
    //	(*client_handler)( client.cSocketFd );
	CommRequestHandler * client_handler = tcp_top_client_handler->getInstance( client );
	(*client_handler)();
	closeConnection ( client.cSocketFd );
	ConnectionsInObserver.extract();

	delete client_handler;
}

// ----------------------------------------------------------- //

void TCPServer::interruptClientHandler( int client_sock )
{
}

// ----------------------------------------------------------- //

void TCPServer::checkError(int err, const std::string &err_msg)
{
	if (err < 0)
	{
		std::stringstream err_stream;
		err_stream << err_msg << "\nErrno: " << errno;
		throw new CommError(FailedConnection, err_stream.str() );
	}
}
